package mr.weblog.part1;

import static mr.weblog.support.WeblogHBaseSchema.ENHANCED_LOGS_FAMILY_CONTENT;
import static mr.weblog.support.WeblogHBaseSchema.WEBLOG_CLIENT_IP_COL;
import static mr.weblog.support.WeblogHBaseSchema.WEBLOG_PRICE_COL;
import static mr.weblog.support.WeblogHBaseSchema.WEBLOG_PRICE_TIMESTAMP_COL;
import static mr.weblog.support.WeblogHBaseSchema.WEBLOG_STOCK_URL_COL;
import static mr.weblog.support.WeblogHBaseSchema.WEBLOG_TIMESTAMP_COL;
import static mr.weblog.support.WeblogHBaseSchema.WEBLOG_USER_ID_COL;
import static mr.weblog.support.WeblogHBaseSchema.WEBLOG_STOCK_SYMBOL_COL;
import static org.apache.hadoop.hbase.util.Bytes.toBytes;

import java.io.IOException;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.List;

import mr.weblog.support.EnrichedWeblog;
import mr.weblog.support.TextPair;

import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.io.WritableComparator;
import org.apache.hadoop.mapreduce.Partitioner;
import org.apache.hadoop.mapreduce.Reducer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class WeblogReduceSideJoinReducer extends
		Reducer<TextPair, TextPair, NullWritable, Writable> {
	private final Logger log = LoggerFactory.getLogger(WeblogReduceSideJoinReducer.class);
	
	public final static String SPACE = " ";
	public final static String START_BRACKET = "[";
	public final static String END_BRACKET = "]";
	public final static String QUOTE = "\"";
	
	private final List<EnrichedWeblog> leftSide = new ArrayList<EnrichedWeblog>();
	private final static SimpleDateFormat format = new SimpleDateFormat("dd/MMM/yyyy:hh:mm:ss z");
			
	@Override
	protected void reduce(TextPair tuppleKey, Iterable<TextPair> joined,
			Context context) throws IOException, InterruptedException {
		String valStr = "";
		
		EnrichedWeblog stockEntry = new EnrichedWeblog();
		log.debug("Processing key [{}]", tuppleKey);
		leftSide.clear();
		
		for (TextPair t : joined) {
			Text val = t.getFirst();
			Text indicator = t.getSide();
			valStr = val.toString();
			
			if (indicator.toString().equals("L")) {
				addEnrichedWeblog(valStr);
			} 
			else if (leftSide.isEmpty()) {
				log.debug("No join");
				return;
			} 
			else {
				// create a string that mimics stock price data so
				// we can reuse the comparator
				try{
					stockEntry.setTimestamp(Long.valueOf(valStr.split(",")[0]));
					stockEntry.setPrice(Double.parseDouble(valStr.split(",")[1]));
					enrichWeblog(stockEntry);
				}
				catch(Exception e)
				{
					System.out.println(valStr);
					e.printStackTrace();
				}
			}
		}
		
		//write results into HBase
		for(int i = 0; i<leftSide.size(); i++){	
			String logStr = leftSide.get(i).getLog();
		    Put put = new Put(toBytes(tuppleKey.getFirst() + "_" + leftSide.get(i).getTimestamp().toString()));
		    
		    int beginIndex = 0;
		    int endIndex = logStr.indexOf(SPACE);
	        put.add(ENHANCED_LOGS_FAMILY_CONTENT, WEBLOG_CLIENT_IP_COL, toBytes(logStr.substring(beginIndex, endIndex)));
	        
	        beginIndex = endIndex+1;
	        endIndex = logStr.indexOf(SPACE, beginIndex);
	        put.add(ENHANCED_LOGS_FAMILY_CONTENT, WEBLOG_USER_ID_COL, toBytes(logStr.substring(beginIndex, endIndex)));

	        beginIndex = logStr.indexOf(START_BRACKET, beginIndex)+1;
	        endIndex = logStr.indexOf(END_BRACKET, beginIndex);
	        put.add(ENHANCED_LOGS_FAMILY_CONTENT, WEBLOG_TIMESTAMP_COL, toBytes(logStr.substring(beginIndex, endIndex)));
		    
	        beginIndex = logStr.indexOf(QUOTE, beginIndex)+1;
	        endIndex = logStr.indexOf(QUOTE, beginIndex);
	        put.add(ENHANCED_LOGS_FAMILY_CONTENT, WEBLOG_STOCK_URL_COL, toBytes(logStr.substring(beginIndex, endIndex)));
		    
	        put.add(ENHANCED_LOGS_FAMILY_CONTENT, WEBLOG_PRICE_TIMESTAMP_COL, toBytes(leftSide.get(i).getPrice_timestamp().toString()));
	        put.add(ENHANCED_LOGS_FAMILY_CONTENT, WEBLOG_PRICE_COL, toBytes(Double.toString(leftSide.get(i).getPrice())));
	        put.add(ENHANCED_LOGS_FAMILY_CONTENT, WEBLOG_STOCK_SYMBOL_COL, toBytes(tuppleKey.getFirst().toString()));
	        context.write(NullWritable.get(), put);
		}
	}

	final public static Comparator<EnrichedWeblog> comparator = new Comparator<EnrichedWeblog>() {
		@Override
		public int compare(EnrichedWeblog o1, EnrichedWeblog o2) {
			return o1.getTimestamp().compareTo(o2.getTimestamp());
		}
	};
	
	public void addEnrichedWeblog(String valStr){
		EnrichedWeblog enrichedLog = new EnrichedWeblog();
		enrichedLog.setLog(valStr);
		enrichedLog.setTimestamp(getTimestampStr(valStr));
		//add the new stock data to the list sorted
		int index = Collections.binarySearch(leftSide, enrichedLog, comparator);
		if(index<0) 
			index=Math.abs(index)-1;
		leftSide.add(index, enrichedLog);
	}
	
	public void enrichWeblog(EnrichedWeblog stockEntry){
		int index = Collections.binarySearch(leftSide, stockEntry, comparator);
		if (index < 0)
			index = Math.abs(index) - 1;
		
		//enriched the stocks
		for(int i = index; i<leftSide.size(); i++){
			if(leftSide.get(i).getPrice_timestamp().compareTo(stockEntry.getTimestamp()) > 0l)
				break;
			
			leftSide.get(i).setPrice_timestamp(stockEntry.getTimestamp());
			leftSide.get(i).setPrice(stockEntry.getPrice());
		}
	}
	
	//assume web log is well formed
	public static long getTimestampStr(String str){

		try {
			int beginPos = str.indexOf('[');
			int endStr = str.indexOf(']');
			return format.parse(str.substring(beginPos+1, endStr)).getTime();
			
		} catch (Exception e1) {
			// TODO Auto-generated catch block
			System.out.println(str);
			e1.printStackTrace();
		}
		return 0;
	}
	
	public static class ReduceSidePartitioner extends
			Partitioner<TextPair, TextPair> {
		@Override
		public int getPartition(TextPair key, TextPair TextPair, int numPartitions) {
			return (key.getFirst().hashCode() & Integer.MAX_VALUE) % numPartitions;
		}
	}

	public static class ReduceSideGroupComparator extends WritableComparator {
		private final Logger log = LoggerFactory
				.getLogger(ReduceSideGroupComparator.class);

		protected ReduceSideGroupComparator() {
			super(TextPair.class, true);
		}

		@Override
		public int compare(WritableComparable a, WritableComparable b) {
			TextPair t1 = (TextPair) a;
			TextPair t2 = (TextPair) b;
			int res = t1.getFirst().compareTo(t2.getFirst());
			log.debug(t1 + " <> " + t2 + " = " + res);
			return res;
		}
	}
}
